package com.j.lemon.learn.flink.datastream;

import com.j.lemon.learn.flink.WordCount;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * union、connect、comap、coflatmap
 * @author lijunjun
 */
public class UnionOperator {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<WordCount> dataStream = env.fromElements(
                new WordCount("zhangsan", 1),
                new WordCount("lisi", 2),
                new WordCount("lisi", 3));

        DataStream<Tuple3> tuple3DataStream = env.fromElements(
                new Tuple3(1,2,3),
                new Tuple3(4,5,6)
        );

        dataStream.join(tuple3DataStream).where(new KeySelector<WordCount, String>() {
            @Override
            public String getKey(WordCount wordCount) throws Exception {
                return wordCount.getWord();
            }
        });


        ConnectedStreams<WordCount, Tuple3> connect =
                dataStream.connect(tuple3DataStream);

        connect.map(new CoMapFunction<WordCount, Tuple3, Object>() {
            @Override
            public Object map1(WordCount wordCount) throws Exception {
                return null;
            }

            @Override
            public Object map2(Tuple3 tuple3) throws Exception {
                return null;
            }
        });


    }
}
